using JGSY.CMS.LowCode.Platform.Application.Interfaces;
using System;
using System.Threading.Tasks;
using System.Net.Sockets;
using System.Text;
using System.Collections.Concurrent;
using System.Security.Cryptography;

namespace JGSY.CMS.LowCode.Platform.Infrastructure.Services
{
    /// <summary>
    /// Memcached 分布式缓存服务实现
    /// 基于 Memcached 协议的分布式缓存服务，支持一致性哈希、连接池、故障转移等高级功能
    /// </summary>
    public class MemcachedCacheService : ICacheService
    {
        /// <summary>
        /// 日志记录器
        /// </summary>
        private readonly ILogger<MemcachedCacheService> _logger;
        /// <summary>
        /// 缓存命中次数
        /// </summary>
        private long _hitCount = 0;
        /// <summary>
        /// 缓存未命中次数
        /// </summary>
        private long _missCount = 0;
        /// <summary>
        /// 故障转移次数（主机切换）
        /// </summary>
        private long _failoverCount = 0;
        /// <summary>
        /// 故障转移异步通知事件（参数：主机、key）
        /// </summary>
        public event Action<string, string>? OnCacheFailover;
        /// <summary>
        /// 获取缓存统计信息（命中、未命中、故障转移）
        /// </summary>
        public (long hit, long miss, long failover) GetCacheStats() => (_hitCount, _missCount, _failoverCount);
        /// <summary>
        /// 一致性哈希虚拟节点映射表（分布式分片）
        /// </summary>
        private readonly SortedDictionary<uint, (string host, int port)> _hashRing = new();
        /// <summary>
        /// 每台服务器虚拟节点数量（提升哈希均匀性）
        /// </summary>
        private const int _virtualNodesPerServer = 100;
        /// <summary>
        /// 最大重试次数（连接池获取连接）
        /// </summary>
    private const int _maxRetry = 3;
        /// <summary>
        /// 重试间隔（毫秒）
        /// </summary>
    private const int _retryDelayMs = 200;
        /// <summary>
        /// 连接池最大容量
        /// </summary>
    private const int _maxPoolSize = 20;
        /// <summary>
        /// 健康检测间隔（毫秒）
        /// </summary>
        private readonly int _healthCheckIntervalMs = 10000;
        /// <summary>
        /// 每台服务器连接池初始容量
        /// </summary>
        private readonly int _poolSize = 5;
        /// <summary>
        /// 健康检测定时器
        /// </summary>
        private readonly Timer _healthCheckTimer;
        /// <summary>
        /// 服务器列表（分布式节点）
        /// </summary>
        private readonly List<(string host, int port)> _servers;
        /// <summary>
        /// 连接池字典（每台服务器独立池）
        /// </summary>
        private readonly ConcurrentDictionary<string, ConcurrentBag<TcpClient>> _pools = new();
        /// <summary>
        /// 连接超时时间（毫秒），从配置注入
        /// </summary>
        private readonly int _connectionTimeoutMs = 5000;

        /// <summary>
        /// 初始化 Memcached 缓存服务实例
        /// 构建连接池、初始化一致性哈希环，并启动健康检查定时器
        /// </summary>
        /// <param name="servers">Memcached 服务器列表</param>
        /// <param name="logger">日志记录器</param>
        /// <param name="connectionTimeoutMs">连接超时时间（毫秒）</param>
        public MemcachedCacheService(List<(string host, int port)> servers, ILogger<MemcachedCacheService> logger, int connectionTimeoutMs = 5000)
        {
            _servers = servers;
            _logger = logger;
            _connectionTimeoutMs = connectionTimeoutMs;
            foreach (var s in servers)
            {
                var key = s.host + ":" + s.port;
                var bag = new ConcurrentBag<TcpClient>();
                // 初始化连接池
                for (int i = 0; i < _poolSize; i++) bag.Add(new TcpClient());
                _pools[key] = bag;
                // 初始化虚拟节点哈希环
                for (int v = 0; v < _virtualNodesPerServer; v++)
                {
                    var vnodeKey = key + "#" + v;
                    var hash = BitConverter.ToUInt32(SHA256.HashData(Encoding.UTF8.GetBytes(vnodeKey)), 0);
                    _hashRing[hash] = s;
                }
            }
            // 启动健康检测定时器，定期维护连接池
            _healthCheckTimer = new Timer(_ => HealthCheckAllPools(), null, _healthCheckIntervalMs, _healthCheckIntervalMs);
        }

        /// <summary>
        /// 健康检测所有连接池，自动重连、动态扩容，故障转移统计
        /// </summary>
        private void HealthCheckAllPools()
        {
            foreach (var server in _servers)
            {
                var key = server.host + ":" + server.port;
                if (!_pools.TryGetValue(key, out var pool)) continue;
                int healthy = 0, unhealthy = 0;
                var clients = pool.ToArray();
            foreach (var client in clients)
            {
                try
                {
                    if (client.Connected)
                    {
                        healthy++;
                    }
                    else
                    {
                        unhealthy++;
                        // 尝试重连
                        try
                        {
                            client.Close();
                            var newClient = new TcpClient();
                            var connectTask = newClient.ConnectAsync(server.host, server.port);
                            if (!connectTask.Wait(_connectionTimeoutMs))
                            {
                                _failoverCount++;
                                OnCacheFailover?.Invoke(key, "HEALTHCHECK");
                            }
                            else
                            {
                                pool.Add(newClient);
                            }
                        }
                        catch (Exception ex)
                        {
                            _logger?.LogError(ex, $"[HealthCheck] 连接重连异常: {key}");
                            _failoverCount++;
                            OnCacheFailover?.Invoke(key, "HEALTHCHECK");
                        }
                    }
                }
                catch (Exception ex)
                {
                    _logger?.LogError(ex, $"[HealthCheck] 连接池异常: {key}");
                    unhealthy++;
                    _failoverCount++;
                    OnCacheFailover?.Invoke(key, "HEALTHCHECK");
                }
            }
                // 动态扩容：如健康连接不足则补充
                int needAdd = _poolSize - healthy;
                for (int i = 0; i < needAdd; i++)
                {
                    try
                    {
                        var newClient = new TcpClient();
                        var connectTask = newClient.ConnectAsync(server.host, server.port);
                        if (connectTask.Wait(_connectionTimeoutMs))
                        {
                            pool.Add(newClient);
                        }
                        else
                        {
                            _failoverCount++;
                            OnCacheFailover?.Invoke(key, "HEALTHCHECK");
                        }
                    }
                    catch
                    {
                        _failoverCount++;
                        OnCacheFailover?.Invoke(key, "HEALTHCHECK");
                    }
                }
            }
        }
        // ...existing code...
        // ...existing code...
        // 可在此继续扩展缓存操作、批量接口、监控等方法
        /// <summary>
        /// 批量获取缓存（支持分布式分片）
        /// </summary>
        public Dictionary<string, string?> GetMany(IEnumerable<string> keys)
        {
            var result = new Dictionary<string, string?>();
            foreach (var key in keys)
            {
                var value = Get(key);
                result[key] = value;
            }
            return result;
        }

        /// <summary>
        /// 批量设置缓存（支持分布式分片）
        /// </summary>
        public void SetMany(Dictionary<string, string> keyValues, int expireSeconds = 0)
        {
            foreach (var kv in keyValues)
            {
                Set(kv.Key, kv.Value, expireSeconds);
            }
        }

        /// <summary>
        /// 获取当前所有连接池健康状态（监控接口）
        /// </summary>
        public Dictionary<string, (int healthy, int unhealthy)> GetPoolHealthStatus()
        {
            var status = new Dictionary<string, (int healthy, int unhealthy)>();
            foreach (var server in _servers)
            {
                var key = server.host + ":" + server.port;
                if (!_pools.TryGetValue(key, out var pool)) continue;
                int healthy = 0, unhealthy = 0;
                foreach (var client in pool)
                {
                    if (client.Connected) healthy++;
                    else unhealthy++;
                }
                status[key] = (healthy, unhealthy);
            }
            return status;
        }

        /// <summary>
        /// 获取缓存命中率（监控接口）
        /// </summary>
        public double GetCacheHitRate()
        {
            var total = _hitCount + _missCount;
            return total == 0 ? 0 : (double)_hitCount / total;
        }

        /// <summary>
        /// 单个获取缓存（示例，需根据实际协议实现）
        /// </summary>
        public string? Get(string key)
        {
            // 一致性哈希分片，选择目标服务器
            var hash = BitConverter.ToUInt32(SHA256.HashData(Encoding.UTF8.GetBytes(key)), 0);
            var target = _hashRing.Count == 0 ? default : _hashRing[_hashRing.Keys.FirstOrDefault(h => h >= hash)];
            var serverKey = target.host + ":" + target.port;
            if (!_pools.TryGetValue(serverKey, out var pool)) { _missCount++; return null; }
            TcpClient? client = null;
            foreach (var c in pool)
            {
                if (c.Connected) { client = c; break; }
            }
            if (client == null) { _missCount++; return null; }
            try
            {
                var stream = client.GetStream();
                var cmd = $"get {key}\r\n";
                var bytes = Encoding.ASCII.GetBytes(cmd);
                stream.Write(bytes, 0, bytes.Length);
                var buffer = new byte[4096];
                int read = stream.Read(buffer, 0, buffer.Length);
                var resp = Encoding.ASCII.GetString(buffer, 0, read);
                if (resp.StartsWith("VALUE"))
                {
                    var lines = resp.Split('\n');
                    if (lines.Length > 1)
                    {
                        _hitCount++;
                        return lines[1].TrimEnd('\r');
                    }
                }
                _missCount++;
                return null;
            }
            catch (Exception ex)
            {
                _logger?.LogError(ex, $"[Get] 读取缓存异常: {key}");
                _missCount++;
                return null;
            }
        }

        /// <summary>
        /// 单个设置缓存（示例，需根据实际协议实现）
        /// </summary>
        public void Set(string key, string value, int expireSeconds = 0)
        {
            // 一致性哈希分片，选择目标服务器
            var hash = BitConverter.ToUInt32(SHA256.HashData(Encoding.UTF8.GetBytes(key)), 0);
            var target = _hashRing.Count == 0 ? default : _hashRing[_hashRing.Keys.FirstOrDefault(h => h >= hash)];
            var serverKey = target.host + ":" + target.port;
            if (!_pools.TryGetValue(serverKey, out var pool)) return;
            TcpClient? client = null;
            foreach (var c in pool)
            {
                if (c.Connected) { client = c; break; }
            }
            if (client == null) return;
            try
            {
                var stream = client.GetStream();
                var cmd = $"set {key} 0 {expireSeconds} {Encoding.UTF8.GetByteCount(value)}\r\n{value}\r\n";
                var bytes = Encoding.ASCII.GetBytes(cmd);
                stream.Write(bytes, 0, bytes.Length);
                var buffer = new byte[256];
                int read = stream.Read(buffer, 0, buffer.Length);
                var resp = Encoding.ASCII.GetString(buffer, 0, read);
                // 可根据 resp 判断是否成功
            }
            catch (Exception ex)
            {
                _logger?.LogError(ex, $"[Set] 写入缓存异常: {key}");
                // 忽略异常，健康检测会自动处理
            }
        }

        /// <summary>
        /// 移除缓存（实现 ICacheService.RemoveAsync）
        /// </summary>
        public async Task RemoveAsync(string key)
        {
            // 一致性哈希分片，选择目标服务器
            uint hash = BitConverter.ToUInt32(SHA256.HashData(Encoding.UTF8.GetBytes(key)), 0);
            (string host, int port) target = _hashRing.Count == 0 ? default : _hashRing[_hashRing.Keys.FirstOrDefault(h => h >= hash)];
            string serverKey = target.host + ":" + target.port;
            if (!_pools.TryGetValue(serverKey, out var pool)) return;
            TcpClient? client = null;
            foreach (var c in pool)
            {
                if (c.Connected) { client = c; break; }
            }
            if (client == null) return;
            try
            {
                NetworkStream stream = client.GetStream();
                string cmd = $"delete {key}\r\n";
                byte[] bytes = Encoding.ASCII.GetBytes(cmd);
                await stream.WriteAsync(bytes, 0, bytes.Length);
                byte[] buffer = new byte[256];
                int read = await stream.ReadAsync(buffer, 0, buffer.Length);
                string resp = Encoding.ASCII.GetString(buffer, 0, read);
                // 可根据 resp 判断是否成功
            }
            catch (Exception ex)
            {
                _logger?.LogError(ex, $"[Remove] 删除缓存异常: {key}");
                // 忽略异常，健康检测会自动处理
            }
        }

        /// <summary>
        /// 异步设置缓存项
        /// 将键值对存储到 Memcached 缓存中，支持自定义过期时间
        /// </summary>
        /// <typeparam name="T">缓存值的类型</typeparam>
        /// <param name="key">缓存键</param>
        /// <param name="value">缓存值</param>
        /// <param name="ttl">过期时间，默认30分钟</param>
        /// <returns>设置操作的异步任务</returns>
        public Task SetAsync<T>(string key, T value, TimeSpan? ttl = null)
        {
            try
            {
                var serializedValue = value is string strValue 
                    ? strValue 
                    : System.Text.Json.JsonSerializer.Serialize(value);
                
                var expireSeconds = (int)(ttl?.TotalSeconds ?? 1800); // 默认30分钟
                
                // 调用同步Set方法
                Set(key, serializedValue, expireSeconds);
                
                return Task.CompletedTask;
            }
            catch (Exception ex)
            {
                _logger?.LogError(ex, $"[SetAsync] 异步设置缓存失败: {key}");
                return Task.FromException(ex);
            }
        }

        /// <summary>
        /// 异步获取缓存项
        /// 从 Memcached 缓存中检索指定键的值并反序列化
        /// </summary>
        /// <typeparam name="T">缓存值的类型</typeparam>
        /// <param name="key">缓存键</param>
        /// <returns>缓存值，如果不存在则返回默认值</returns>
        public Task<T?> GetAsync<T>(string key)
        {
            try
            {
                // 调用同步Get方法
                var result = Get(key);
                
                if (result == null)
                    return Task.FromResult(default(T));

                if (typeof(T) == typeof(string))
                {
                    return Task.FromResult((T?)(object)result);
                }

                // 反序列化JSON
                var deserializedValue = System.Text.Json.JsonSerializer.Deserialize<T>(result);
                return Task.FromResult(deserializedValue);
            }
            catch (Exception ex)
            {
                _logger?.LogError(ex, $"[GetAsync] 异步获取缓存失败: {key}");
                return Task.FromResult(default(T));
            }
        }
    }
}
